package com.test.demo.transaction2;

import com.alibaba.fastjson.JSON;
import com.test.demo.dao.entity.TransferRecord;
import com.test.demo.utils.HashUtil;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 事务类
 * @author Gao
 * @Component 最普通的组件，可以被注入到spring容器进行管理
 * @Repository 作用于持久层
 * @Service 作用于业务逻辑层
 * @Controller 作用于表现层（spring-mvc的注解）
 * @Component 是一个通用的Spring容器管理的单例bean组件。
 * 而@Repository, @Service, @Controller就是针对不同的使用场景所采取的特定功能化的注解组件。
 * 因此，当你的一个类被@Component所注解，那么就意味着同样可以用@Repository, @Service, @Controller来替代它，
 */
@Component
public class TransactionListenerImpl2 implements TransactionListener {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * @Autowired //默认按type注入
     * @Qualifier("businessService") //一般作为@Autowired()的修饰用
     * @Resource(name="businessService") //默认按name注入，可以通过name和type属性进行选择性注入
     * 一般@Autowired和@Qualifier一起用，@Resource单独用。
     * 当然没有冲突的话@Autowired也可以单独用
     */
    @Resource
    private BusinessService businessService;

    /**
     * Message中封装了事务id String transactionId;
     * @param msg
     * @param arg
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        logger.info("prepare消息执行成功，执行本地事务，message:{}", msg);
        TransferRecord transferRecord = JSON.parseObject(msg.getBody(), TransferRecord.class);
        LocalTransactionState state = LocalTransactionState.UNKNOW;
        try {
            boolean isCommit = businessService.doTransfer(transferRecord.getFromUserId(), transferRecord.getToUserId()
                    , transferRecord.getChangeMoney(), transferRecord.getRecordNo(), msg.getTransactionId());
            if (isCommit) {
                logger.info("[发送事务消息]本地事务执行成功！");
                state = LocalTransactionState.COMMIT_MESSAGE;
            } else {
                logger.info("[发送事务消息]本地事务执行失败！");
                state = LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            System.out.println("转账失败,fromUserId:" + transferRecord.getFromUserId()
                    + ",toUserId:" + transferRecord.getToUserId() + ",money:" + transferRecord.getChangeMoney());
            logger.warn("error:{}", e);
        }
        return state;
    }

    /**
     * 本地事务回查，当未收到本地事务执行结果
     * 没有收到COMMIT_MESSAGE或ROLLBACK_MESSAGE的时候执行的
     * 如果LocalTransactionState state 还是 UNKNOW 或
     * 者本地事务异常会执行本地事务回查 throw new Exception("666");
     *
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        logger.info("本地事务状态回查msg:{}", msg);
        //消息ID（有可能消息体一样，但消息ID不一样，当前消息属于半事务消息，所以消息ID在消息队列RocketMQ版控制台无法查询）。
        String msgId = msg.getMsgId();
        //消息体内容进行crc32，也可以使用其它的方法如MD5。
        long crc32Id = HashUtil.crc32Code(msg.getBody());
        //消息ID和crc32Id主要是用来防止消息重复。
        //如果业务本身是幂等的，可以忽略，否则需要利用msgId或crc32Id来做幂等。
        //如果要求消息绝对不重复，推荐做法是对消息体使用crc32或MD5来防止重复消息。
        //业务自己的参数对象，这里只是一个示例，需要您根据实际情况来处理。
        // 回查可以做消息幂等性去重
        LocalTransactionState state = LocalTransactionState.UNKNOW;
        try {
            boolean isCommit = businessService.checkTransferStatus(msg.getTransactionId());
            if (isCommit) {
                state = LocalTransactionState.COMMIT_MESSAGE;
            } else {
                state = LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            logger.warn("error:{}", e);
        }
        return state;
    }
}
